package cn.com.ry.framework.mongodb.sync.service;

import cn.com.ry.framework.mongodb.sync.entity.Page;
import cn.com.ry.framework.mongodb.sync.entity.PointBean;
import cn.com.ry.framework.mongodb.sync.util.SynLogUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.ValueFilter;
import org.bson.types.ObjectId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.regex.Pattern;

@Service
public class SynService {
    private static final Logger logger = LoggerFactory.getLogger(SynService.class);
    @Autowired
    private SynLogUtil synLogUtil;
    @Autowired
    private MongoTemplate mongoTemplate;
    @Autowired
    private JdbcTemplate jdbcTemplate;

    //线程数
    @Value("${syn.consuer.threadnum:20}")
    private int THREAD_NUM = 20;
    //需要跳过的表
    @Value("${syn.skip.collection}")
    private String skipCollection;
    //状态,如果为true 则跳过 skipCollection 内容，如果 false，则只同步skipCollection内容
    @Value("${syn.skip.status:true}")
    private boolean status;
    @Value("${syn.point.flag:false}")
    private boolean pointFlag;
    @Value("${syn.collection.log.path}")
    private String collectionLogPath;
    @Value("${syn.other.collections:}")
    private String otherCollections;

    /**
     * 获取collection列表
     *
     * @return
     */
    public Set<String> getCollectionNames() {
        return mongoTemplate.getCollectionNames();
    }

    public void createOtherTable() {
        String[] otherCollectionStr = otherCollections.split(",");
        if (otherCollectionStr != null && otherCollectionStr.length > 0) {
            for (int i = 0; i < otherCollectionStr.length; i++) {
                createTable(otherCollectionStr[i]);
            }
        }
    }

    /**
     * 获取collection总数量
     *
     * @param collectionName
     * @return
     */
    public Page getCollectionCount(String collectionName, int pageSize) {
        Page page = new Page();
        if (StringUtils.isEmpty(collectionName)) {
            return page;
        }
        //判断如果是断点续传，加载上传 objectId
        String maxObjectId = null;
        if (pointFlag) {
            maxObjectId = synLogUtil.getPropertyValue(collectionLogPath, collectionName, "");
        }

        //如果maxObjectId为空，则获取最大_id
        if (maxObjectId == null || maxObjectId.trim().equals("")) {
            Query query = new Query().with(new Sort(new Sort.Order(Sort.Direction.DESC, "_id"))).limit(1).skip(0);
            //serialnum 特殊处理
            if (collectionName.equalsIgnoreCase("serialnum")) {
                Pattern pattern = Pattern.compile("20[1-9][0-9]");
                query.addCriteria(new Criteria().orOperator(Criteria.where("key").not().regex(pattern), Criteria.where("key").is("ZPB20{SN:4}_2020")));
            }
            Map resultMap = mongoTemplate.findOne(query, Map.class, collectionName);
            if (resultMap != null && resultMap.get("_id") instanceof ObjectId) {
                maxObjectId = resultMap.get("_id").toString();
            }
        }

        page.setMaxObjectId(maxObjectId);


        long count = 0;
        //序列号特殊处理
        Query query = new Query();
        if (collectionName.equalsIgnoreCase("serialnum")) {
            Pattern pattern = Pattern.compile("20[1-9][0-9]");
            query.addCriteria(new Criteria().orOperator(Criteria.where("key").not().regex(pattern), Criteria.where("key").is("ZPB20{SN:4}_2020")));
        }
        //如果_id类型不为ObjectId，则全部查出
        if (maxObjectId != null && !maxObjectId.trim().equals("")) {
            ObjectId objectId = new ObjectId(maxObjectId);
            //根据最大_id返回条数
            query.addCriteria(new Criteria("_id").lte(objectId));

            count = mongoTemplate.count(query, collectionName);
        } else {
            count = mongoTemplate.count(query, collectionName);
        }
        page.setCollectionName(collectionName);
        page.setTotalRecord((int) count);
        page.setPageSize(pageSize);
        return page;
    }

    /**
     * 获取mongodb中的数据
     *
     * @param page
     * @return
     */
    public synchronized List<String> getCollectionData(Page page) {
        List<String> result = new ArrayList<>();
        int start = page.getPageSize() * (page.getCurrentPage() - 1);
        Query query = new Query().limit(page.getPageSize()).skip(start).with(new Sort(new Sort.Order(Sort.Direction.ASC, "_id")));
        //序列号特殊处理
        String collectionName = page.getCollectionName();
        if (collectionName.equalsIgnoreCase("serialnum")) {
            Pattern pattern = Pattern.compile("20[1-9][0-9]");
            query.addCriteria(new Criteria().orOperator(Criteria.where("key").not().regex(pattern), Criteria.where("key").is("ZPB20{SN:4}_2020")));
        }
        String objectId = page.getMaxObjectId();
        if (objectId != null && !objectId.trim().equals("")) {
            query.addCriteria(new Criteria("_id").lte(new ObjectId(objectId)));
        }
        List<Map> resultList = mongoTemplate.find(query, Map.class, page.getCollectionName());
        //将对象转成string
        if (resultList != null && resultList.size() > 0) {
            for (int i = 0; i < resultList.size(); i++) {
                try {
                    Map resultMap = resultList.get(i);
                    //objectId转换
                    ValueFilter filter = new ValueFilter() {
                        @Override
                        public Object process(Object obj, String s, Object v) {
                            logger.debug(s);
                            if (v instanceof ObjectId) {
                                return v.toString();
                            } else if (v instanceof Date) {
                                return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(v);
                            }
                            return v;
                        }
                    };
                    String dataJson = JSON.toJSONString(resultMap, filter);
                    result.add(dataJson);
                } catch (Exception e) {
                    logger.error("格式转换失败", e);
                    continue;
                }
            }
        }
        return result;
    }

    /**
     * 创建表
     *
     * @param table
     */
    @Transactional(propagation = Propagation.REQUIRED)
    public void createTable(String table) {
        String droptableSql = "DROP TABLE IF EXISTS `" + table + "`;\n";
        String createTableSql = "";

        if (table.equalsIgnoreCase("serialnum")) {
            createTableSql = "CREATE TABLE `" + table + "`  (\n" +
                    "  `id` bigint(20) NOT NULL AUTO_INCREMENT,\n" +
                    "  `datajson` json NULL,\n" +
                    "  `_id` VARCHAR(200) GENERATED ALWAYS AS (`datajson` ->> '$._id') NOT NULL,\n" +
                    "  `key` VARCHAR(200) GENERATED ALWAYS AS (`datajson` ->> '$.key'),\n" +
                    "  PRIMARY KEY (`id`)\n" +
                    ") ENGINE = InnoDB AUTO_INCREMENT = 1";
        } else if (table.equalsIgnoreCase("mqmes")) {
            createTableSql = "CREATE TABLE `" + table + "`  (\n" +
                    "  `id` bigint(20) NOT NULL AUTO_INCREMENT,\n" +
                    "  `datajson` json NULL,\n" +
                    "  `_id` VARCHAR(200) GENERATED ALWAYS AS (`datajson` ->> '$._id') NOT NULL,\n" +
                    "  `date` VARCHAR(200) GENERATED ALWAYS AS (`datajson` ->> '$.date'),\n" +
                    "  `mesId` VARCHAR(200) GENERATED ALWAYS AS (`datajson` ->> '$.mesId'),\n" +
                    "  PRIMARY KEY (`id`)\n" +
                    ") ENGINE = InnoDB AUTO_INCREMENT = 1";

        } else if (table.equalsIgnoreCase("timerloginfo")) {
            createTableSql = "CREATE TABLE `" + table + "`  (\n" +
                    "  `id` bigint(20) NOT NULL AUTO_INCREMENT,\n" +
                    "  `datajson` json NULL,\n" +
                    "  `_id` VARCHAR(200) GENERATED ALWAYS AS (`datajson` ->> '$._id') NOT NULL,\n" +
                    "  `endTime` VARCHAR(200) GENERATED ALWAYS AS (`datajson` ->> '$.endTime') ,\n" +
                    "  `timerCode` VARCHAR(200) GENERATED ALWAYS AS (`datajson` ->> '$.timerCode'),\n" +
                    "  PRIMARY KEY (`id`)\n" +
                    ") ENGINE = InnoDB AUTO_INCREMENT = 1";

        } else {
            createTableSql = "CREATE TABLE `" + table + "`  (\n" +
                    "  `id` bigint(20) NOT NULL AUTO_INCREMENT,\n" +
                    "  `datajson` json NULL,\n" +
                    "  `_id` VARCHAR(200) GENERATED ALWAYS AS (`datajson` ->> '$._id') NOT NULL,\n" +
                    "  PRIMARY KEY (`id`)\n" +
                    ") ENGINE = InnoDB AUTO_INCREMENT = 1";
        }
        logger.info("[开始]创建表，执行sql:" + createTableSql);
        jdbcTemplate.update(droptableSql);
        jdbcTemplate.update(createTableSql);
        logger.info("[结束]创建表，执行sql:" + createTableSql);
    }

    /**
     * 创建索引
     */
    public void createIndex() {
        Set<String> collections = getCollectionNames();
        if (collections != null) {
            Iterator it = collections.iterator();
            for (; it.hasNext(); ) {
                String table = (String) it.next();
                String sql = "";
                if (table.equalsIgnoreCase("serialnum")) {
                    sql = "ALTER TABLE `serialnum` ADD UNIQUE INDEX `_serialnum_index_id`(`_id`) USING BTREE;" +
                            "ALTER TABLE `serialnum` ADD INDEX `_serialnum_index_key`(`key`) USING BTREE;";
                } else if (table.equalsIgnoreCase("mqmes")) {
                    sql = "ALTER TABLE `mqmes` ADD INDEX `_mqmes_index_id`(`_id`) USING BTREE;" +
                            "ALTER TABLE `mqmes` ADD INDEX `_mqmes_index_date`(`date`) USING BTREE;" +
                            "ALTER TABLE `mqmes` ADD  INDEX `_mqmes_index_mesId`(`mesId`) USING BTREE";
                } else if (table.equalsIgnoreCase("timerloginfo")) {
                    sql = "ALTER TABLE `timerloginfo` ADD INDEX `_timerloginfo_index_id`(`_id`) USING BTREE;" +
                            "ALTER TABLE `timerloginfo` ADD INDEX `_timerloginfo_index_endTime`(`endTime`) USING BTREE;" +
                            "ALTER TABLE `timerloginfo` ADD INDEX `_timerloginfo_index_timerCode`(`timerCode`) USING BTREE";
                } else {
                    sql = "ALTER TABLE `" + table + "` ADD INDEX `_" + table + "_index_id`(`_id`) USING BTREE;";
                }
                logger.info("[开始]创建索引，执行sql:" + sql);
                try{
                    jdbcTemplate.batchUpdate(sql);
                }catch (Exception e){

                }
                logger.info("[结束]创建索引，执行sql:" + sql);
            }
        }
    }

    public void addTable(String table) {
        String createTableSql = "";

        if (table.equalsIgnoreCase("serialnum")) {
            createTableSql = "CREATE TABLE `" + table + "`  (\n" +
                    "  `id` bigint(20) NOT NULL AUTO_INCREMENT,\n" +
                    "  `datajson` json NULL,\n" +
                    "  `_id` VARCHAR(200) GENERATED ALWAYS AS (`datajson` ->> '$._id') NOT NULL,\n" +
                    "  `key` VARCHAR(200) GENERATED ALWAYS AS (`datajson` ->> '$.key'),\n" +
                    "  PRIMARY KEY (`id`)\n" +
                    ") ENGINE = InnoDB AUTO_INCREMENT = 1";
        } else if (table.equalsIgnoreCase("mqmes")) {
            createTableSql = "CREATE TABLE `" + table + "`  (\n" +
                    "  `id` bigint(20) NOT NULL AUTO_INCREMENT,\n" +
                    "  `datajson` json NULL,\n" +
                    "  `_id` VARCHAR(200) GENERATED ALWAYS AS (`datajson` ->> '$._id') NOT NULL,\n" +
                    "  `date` VARCHAR(200) GENERATED ALWAYS AS (`datajson` ->> '$.date'),\n" +
                    "  `mesId` VARCHAR(200) GENERATED ALWAYS AS (`datajson` ->> '$.mesId'),\n" +
                    "  PRIMARY KEY (`id`)\n" +
                    ") ENGINE = InnoDB AUTO_INCREMENT = 1";

        } else if (table.equalsIgnoreCase("timerloginfo")) {
            createTableSql = "CREATE TABLE `" + table + "`  (\n" +
                    "  `id` bigint(20) NOT NULL AUTO_INCREMENT,\n" +
                    "  `datajson` json NULL,\n" +
                    "  `_id` VARCHAR(200) GENERATED ALWAYS AS (`datajson` ->> '$._id') NOT NULL,\n" +
                    "  `endTime` VARCHAR(200) GENERATED ALWAYS AS (`datajson` ->> '$.endTime'),\n" +
                    "  `timerCode` VARCHAR(200) GENERATED ALWAYS AS (`datajson` ->> '$.timerCode'),\n" +
                    "  PRIMARY KEY (`id`)\n" +
                    ") ENGINE = InnoDB AUTO_INCREMENT = 1";

        } else {
            createTableSql = "CREATE TABLE `" + table + "`  (\n" +
                    "  `id` bigint(20) NOT NULL AUTO_INCREMENT,\n" +
                    "  `datajson` json NULL,\n" +
                    "  `_id` VARCHAR(200) GENERATED ALWAYS AS (`datajson` ->> '$._id') NOT NULL,\n" +
                    "  PRIMARY KEY (`id`)\n" +
                    ") ENGINE = InnoDB AUTO_INCREMENT = 1";
        }
        logger.info("[开始]创建表，执行sql:" + createTableSql);
        jdbcTemplate.update(createTableSql);
        logger.info("[结束]创建表，执行sql:" + createTableSql);
    }

    /**
     * 插入到mysql
     *
     * @param tableName
     * @param dataJsons
     * @return
     */
    public int[] addDataForMysql(String tableName, List<String> dataJsons) {
        String sql = "insert into `" + tableName + "` (datajson) value(?)";
        if (dataJsons == null || dataJsons.size() <= 0) {
            return null;
        }
        List<Object[]> batchArgs = new ArrayList<Object[]>();
        for (int i = 0; i < dataJsons.size(); i++) {
            batchArgs.add(new Object[]{dataJsons.get(i)});
        }
        int[] result = jdbcTemplate.batchUpdate(sql, batchArgs);
        return result;
    }

    /**
     * 同步数据
     *
     * @param collectionName
     */
    public void doSynData(String collectionName) {
        //获取日志，进度断点续传
        String pointPage = synLogUtil.getPropertyValue(collectionName);
        //创建新表
        if (pointPage == null) {
            //创建table   currentPage
            createTable(collectionName);
            pointPage = "0";
        }
        //开启断点续传
        int pointPageInt = Integer.parseInt(pointPage) + 1;
        logger.info("载入断点【" + collectionName + "=" + pointPageInt + "】");
        //获取页数
        Page page = getCollectionCount(collectionName, 1000);
        int totalPage = page.getTotalPage();

        //分页执行,从断点开始
        for (int i = pointPageInt; i < totalPage + 1; i++) {
            long startTime = System.currentTimeMillis();
            logger.info("[开始]同步集合【" + collectionName + "】的数据，进度" + i + "/" + totalPage);
            page.setCurrentPage(i);
            //获取mongodb数据
            List<String> dataJson = getCollectionData(page);
            //存入mysql
            int[] status = addDataForMysql(collectionName, dataJson);
            //记录断点
            if (status != null) {
                synLogUtil.save(collectionName, String.valueOf(i));
                logger.info("记录断点【" + collectionName + "=" + i + "】");
            }
            long endTime = System.currentTimeMillis();
            long totalTime = (endTime - startTime) / 1000;
            long needTime = (totalPage - i) * totalTime;
            logger.info("[开始]同步集合【" + collectionName + "】的数据，进度" + i + "/" + totalPage + ",耗时" + totalTime + "秒，预计还需要" + needTime + "秒");
        }
    }


    /**
     * 断点续传的时候判断是否已经同步
     *
     * @param page
     * @return
     */
    public boolean hasSyn(Page page) {
        logger.info("开始检查第" + page.getQueueSize() + "个任务是否已经处理");
        int start = page.getPageSize() * (page.getCurrentPage() - 1);
        Query query = new Query().limit(page.getPageSize()).skip(start);
        List<Map> resultList = mongoTemplate.find(query, Map.class, page.getCollectionName());
        //将对象转成string
        String id = "";
        if (resultList != null && resultList.size() > 0) {
            Map resultMap = resultList.get(0);
            //objectId转换
            ObjectId objectId = (ObjectId) resultMap.get("_id");
            id = objectId.toString();
        }
        String sql = "select count(*) as count from " + page.getCollectionName() + " where _id=?";
        Map queryMap = jdbcTemplate.queryForMap(sql, new Object[]{id});
        if (queryMap != null && (long) queryMap.get("count") > 0) {
            logger.info("第" + page.getQueueSize() + "个任务已经执行过，则跳过");
            return true;
        } else {
            logger.info("第" + page.getQueueSize() + "个任务未执行，则继续执行");
            return false;
        }
    }

    /**
     * TODO 多线程-生产任务
     *
     * @param blockingQueue
     */
    public void addTask(BlockingQueue blockingQueue) {
        Set<String> collectionNames = getCollectionNames();

        if (collectionNames != null) {
            Iterator<String> it = collectionNames.iterator();
            int queueSize = 0;
            //加载断点
            PointBean pointBean = new PointBean();
            pointBean.setMaxPoint(0);
            pointBean.setMinPoint(0);
            pointBean.setLastPointList(new ArrayList());
            if (pointFlag) {
                pointBean = synLogUtil.getPoint();
                logger.info("断点位置，minPoint=" + pointBean.getMinPoint() + ",maxPoint=" + pointBean.getMaxPoint() + ",上次未完成=" + pointBean.getLastPointList().toString());
            }

            for (; it.hasNext(); ) {
                String collectionName = it.next();
                //跳过不需要同步的表
                if (skipCollection != null && !skipCollection.trim().equals("")) {
                    boolean hasCollection = Arrays.asList(skipCollection.split(",")).contains(collectionName);
                    //如果存在
                    if (hasCollection) {
                        if (status) {
                            continue;
                        }
                    } else {
                        if (!status) {
                            continue;
                        }
                    }
                }


                //获取总数
                Page page = getCollectionCount(collectionName, 1000);

                //记录collection的最大objectId
                synLogUtil.save(collectionLogPath, collectionName, page.getMaxObjectId());

                int totalPage = page.getTotalPage();

                //创建table
                if (pointFlag && pointBean.getMaxPoint() > 0) {
                    addTable(collectionName);
                } else {
                    createTable(collectionName);
                }

                for (int j = 1; j < totalPage + 1; j++) {
                    page = getCollectionCount(collectionName, 1000);
                    page.setCurrentPage(j);
                    page.setQueueSize(++queueSize);
                    try {
                        //跳过已经执行的任务
                        if (queueSize > pointBean.getMaxPoint()) {
                            blockingQueue.put(page);
                        } else {
                            //添加上次失败的任务
                            if (queueSize >= pointBean.getMinPoint()) {
                                List<Integer> lastPointList = pointBean.getLastPointList();
                                if (lastPointList.contains(queueSize) && !hasSyn(page)) {
                                    blockingQueue.put(page);
                                }
                            }
                        }
                    } catch (InterruptedException e) {
                        logger.error("任务添加失败", e);
                    }
                    System.out.println(Thread.currentThread().getName() + "【添加任务】，当前任务数 " + blockingQueue.size());
                }
            }
        }
    }


    /**
     * TODO 多线程-执行任务
     *
     * @param blockingQueue
     */
    @Transactional
    public void excuteTask(BlockingQueue blockingQueue) {
        Page page = null;
        try {
            //解决成功状态为修改成功问题
            //记录日志  threadName= queueId,collectionName,page,pageTotal,status
            page = (Page) blockingQueue.take();
            String collectionName = page.getCollectionName();
            //开始
            String logInfo = page.getQueueSize() + "," + collectionName + "," + page.getCurrentPage() + "," + page.getTotalPage();
            synLogUtil.save(Thread.currentThread().getName(), logInfo + ",0");
            //执行任务
            long startTime = System.currentTimeMillis();
            logger.info("[线程]" + Thread.currentThread().getName() + "[开始]同步集合【" + collectionName + "】的数据，进度" + page.getCurrentPage() + "/" + page.getTotalPage() + ",剩余任务数" + +blockingQueue.size());
            //获取mongodb数据
            List<String> dataJson = getCollectionData(page);
            //存入mysql
            int[] status = addDataForMysql(collectionName, dataJson);
            long endTime = System.currentTimeMillis();
            long totalTime = (endTime - startTime) / 1000;
            logger.info("[线程]" + Thread.currentThread().getName() + "[结束]同步集合【" + collectionName + "】的数据，进度" + page.getCurrentPage() + "/" + page.getTotalPage() + ",耗时" + totalTime + "秒,剩余任务数" + blockingQueue.size());
            //结束
            synLogUtil.save(Thread.currentThread().getName(), logInfo + ",1");
        } catch (InterruptedException e) {
            logger.error("[同步异常]" + Thread.currentThread().getName() + "[开始]同步集合【" + page.getCollectionName() + "】的数据，进度" + page.getCurrentPage() + "/" + page.getTotalPage(), e);
        }
    }

    /**
     * TODO 单线程同步数据
     */
    public void synData() {
        //获取collection列表
        Set<String> collectionNames = getCollectionNames();
        int totalCollection = collectionNames.size();
        if (collectionNames != null) {
            Iterator<String> it = collectionNames.iterator();
            for (int i = 1; it.hasNext(); i++) {
                long startTime = System.currentTimeMillis();
                String collectionName = it.next();
                logger.info("[开始]同步集合【" + collectionName + "】,进度" + i + "/" + totalCollection);
                doSynData(collectionName);
                long endTime = System.currentTimeMillis();
                long totalTime = (endTime - startTime) / 1000;
                long needTime = totalTime * (totalCollection - i);
                logger.info("[结束]同步集合【" + collectionName + "】,进度" + i + "/" + totalCollection + "，耗费" + totalTime + "秒,预计还需" + needTime + "秒");

            }
        }
    }
}
